package cn.ztuo.bitrade.consumer;

import cn.ztuo.bitrade.Trade.CoinTrader;
import cn.ztuo.bitrade.Trade.CoinTraderFactory;
import cn.ztuo.bitrade.entity.ContractCoin;
import cn.ztuo.bitrade.entity.ContractOrder;
import cn.ztuo.bitrade.service.ContractCoinService;
import cn.ztuo.bitrade.service.ContractOrderDetailService;
import cn.ztuo.bitrade.service.ContractOrderService;
import cn.ztuo.bitrade.service.ContractWalletService;
import cn.ztuo.bitrade.util.MessageResult;
import cn.ztuo.bitrade.util.RedisUtil;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.List;

@Slf4j
@Component
public class ContractOrderConsumer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @Autowired
    private CoinTraderFactory traderFactory;
    @Autowired
    private ContractOrderService contractOrderService;
    @Autowired
    private ContractCoinService contractCoinService;
    @Autowired
    private RedisUtil redisUtil;



    @KafkaListener(topics = "contract-order",containerFactory = "kafkaListenerContainerFactory")
    public void OrderSubmitted(List<ConsumerRecord<String,String>> records){
        log.info("订单发送到cotract系统  进行撮合匹配=============");
        for (int i = 0; i < records.size(); i++) {
            ConsumerRecord<String,String> record = records.get(i);
            log.info("onOrderSubmitted:topic={},key={}", record.topic(), record.key());
            String symbol = record.key();
            ContractOrder order = JSON.parseObject(record.value(), ContractOrder.class);
            log.info("获取到的order  : "+order);
            if (order == null) {
                return;
            }
            CoinTrader trader = traderFactory.getTrader(symbol);
            //如果当前币种交易暂停会自动取消订单====================起始保证金
            if (trader.isTradingHalt() || !trader.getReady()) {
                //撮合器未准备完成，撤回当前等待的订单
                kafkaTemplate.send("contract-simple-order-cancel-success", order.getSymbol(), JSON.toJSONString(order));
            } else {
                try {
                    long startTick = System.currentTimeMillis();
                    // 撮合处理
                    trader.trade(order,contractOrderService);
                    log.info("complete trade,{}ms used!", System.currentTimeMillis() - startTick);
                } catch (Exception e) {
                    e.printStackTrace();
                    log.error("====交易出错，退回订单===");
                    kafkaTemplate.send("contract-simple-order-cancel-success", order.getSymbol(), JSON.toJSONString(order));
                }
            }
        }
    }

    @KafkaListener(topics = "contract-symbol",containerFactory = "kafkaListenerContainerFactory")
    public void onAddCoinTradeByExchangeCoin(List<ConsumerRecord<String,String>> records){
        for (int i = 0; i < records.size(); i++) {
            ConsumerRecord<String,String> record  = records.get(i);
            log.info("接收contract-product>>topic={},value={},size={}",record.topic(),record.value(),records.size());
            List<ContractCoin> exchangeCoins = JSON.parseArray(record.value(), ContractCoin.class);
            exchangeCoins.forEach(exchangeCoin -> {
                CoinTrader coinTrader = traderFactory.getTrader(exchangeCoin.getSymbol());
                ContractCoin coin = contractCoinService.findBySymbol(exchangeCoin.getSymbol());
                if(coinTrader==null){
                    coinTrader = new CoinTrader(coin,redisUtil);
                    coinTrader.setKafkaTemplate(kafkaTemplate);
                    coinTrader.setBaseCoinScale(exchangeCoin.getBaseCoinScale());
                    coinTrader.setCoinScale(exchangeCoin.getCoinScale());
                    coinTrader.initialize();
                    coinTrader.setReady(true);
                    traderFactory.addTrader(exchangeCoin.getSymbol(),coinTrader);
                    log.info("添加完毕");
                }
            });
        }
    }



//    /**
//     * 处理撮合市价订单
//     *
//     * @param records
//     */
//    @KafkaListener(topics = "contract-market-price-order", containerFactory = "kafkaListenerContainerFactory")
//    public void onOrderSubmitted(List<ConsumerRecord<String, String>> records) throws ParseException {
//        //获取格式化对象
//        NumberFormat nt = NumberFormat.getPercentInstance();
//        //设置百分数精确度2即保留两位小数
//        nt.setMinimumFractionDigits(2);
//        for (int i = 0; i < records.size(); i++) {
//            ConsumerRecord<String, String> record = records.get(i);
//            log.info("onOrderSubmitted:topic={},key={}", record.topic(), record.key());
//            String symbol = record.key();
//            ContractOrder order = JSON.parseObject(record.value(), ContractOrder.class);
//            if (order == null) {
//                return;
//            }
//            ContractCoin bySymbol = contractCoinService.findBySymbol(symbol);
//            Long memberId = order.getMemberId();
//            //获取用户钱包
//           // ContractWallet byCoinIdAndMemberId = contractWalletService.findByCoinIdAndMemberId(bySymbol.getCoinSymbol(), memberId);
//            //获取合约的配置信息
//            //提供流动性的费率
//            String proxidefee = nt.parse(bySymbol.getProvideFee()) + "";
//            //提取流动性的费率
//            String mentionfee = nt.parse(bySymbol.getMentionFee()) + "";
//            if (order.getDirection() == ContractOrderDirection.BUY) {
//                List<ContractOrder> contractOrders = contractOrderService.findByDirectionAndStatus(ContractOrderDirection.SELL, ContractOrderStatus.NODEAL);
//                if (contractOrders != null) {
//                    for (ContractOrder contractOrder : contractOrders) {
//                        //获取匹配用户钱包
//                        //ContractWallet byCoinId = contractWalletService.findByCoinIdAndMemberId(bySymbol.getCoinSymbol(), contractOrder.getMemberId());
//
//                        //匹配的订单手数大于等于
//                        if (contractOrder.getAmount().compareTo(order.getAmount()) > -1) {
//                            contractOrder.setAmount(BigDecimalUtils.sub(contractOrder.getAmount(),order.getAmount()));
//                            //将用户的手数赋值给成交量
//                            order.setTradedAmount(order.getAmount());
//                            ContractOrderDetail contractOrderDetail = new ContractOrderDetail();
//                            //交易的数量
//                            contractOrderDetail.setAmount(order.getTradedAmount());
//                            //买卖的方向
//                            contractOrderDetail.setDirection(order.getDirection());
//                            //用户保证金
//                            BigDecimal marginTrade = order.getStart();
//                            //用户提取流动性的费率  保证金乘以 提取流动性费率
//                            BigDecimal mul1 = BigDecimalUtils.mul(marginTrade, new BigDecimal(mentionfee));
//                            //保证金减去流动性费率
//                            BigDecimal sub = BigDecimalUtils.sub(marginTrade, mul1);
//                            //保证金
//                            contractOrderDetail.setTurnover(sub);
//                            //手续费
//                            contractOrderDetail.setFee(mul1);
//                            //contractOrderDetail.setDirection(order.getDirection());
//                            contractOrderDetail.setTime(Calendar.getInstance().getTimeInMillis());
//                            contractOrderDetail.setOrderId(order.getOrderId());
//                            contractOrderDetail.setMemberId(order.getMemberId());
//                            contractOrderDetail.setPrice(order.getPrice());
//
//                            contractOrderDetailService.save(contractOrderDetail);
//                            contractOrder.setTradedAmount(order.getAmount());
//                            //用户获取提供流动性费率
//                            BigDecimal proxide = BigDecimalUtils.mul(mul1, new BigDecimal(proxidefee));
//                            //可用余额减去保证金  添加冻结余额 添加 奖励流动性费用
//                            contractWalletService.findByBalancedongjie(memberId,sub,proxide);
//                            //奖励匹配用户
//                            //byCoinId.setBalance(BigDecimalUtils.add(proxide, byCoinId.getBalance()));
//                            log.info("计算匹配者手续费====================");
//                            //获取开仓价格
//                            BigDecimal price = contractOrder.getPrice();
//                            //匹配者对应的仓位保证金
//                            BigDecimal getbond = ContractUtils.getbond(order.getAmount(), price, order.getStart());
//                            //用户提取流动性的费率  保证金乘以 提取流动性费率
//                            BigDecimal getliud = BigDecimalUtils.mul(getbond, new BigDecimal(mentionfee));
//                            //保证金减去流动性费率
//                            BigDecimal getjian = BigDecimalUtils.sub(getbond, getliud);
//                            //用户获取提供流动性费率
//                            BigDecimal gethuo = BigDecimalUtils.mul(getliud, new BigDecimal(proxidefee));
//                            //可用余额减去保证金  添加冻结余额  添加 奖励流动性费用
//                            contractWalletService.findByBalancedongjie(contractOrder.getMemberId(),getjian,gethuo);
//                            ContractOrderDetail contractOrdersell = new ContractOrderDetail();
//                            contractOrdersell.setPrice(contractOrder.getPrice());
//                            contractOrdersell.setMemberId(contractOrder.getMemberId());
//                            contractOrdersell.setOrderId(contractOrder.getOrderId());
//                            contractOrdersell.setTime(Calendar.getInstance().getTimeInMillis());
//                            contractOrdersell.setDirection(contractOrder.getDirection());
//                            contractOrdersell.setTurnover(getjian);
//                            contractOrdersell.setFee(getliud);
//                            contractOrdersell.setAmount(contractOrder.getAmount());
//                            //contractOrdersell.setMarginTrade();
//                            if (BigDecimalUtils.isEqual(contractOrder.getAmount(),BigDecimal.ZERO)){
//                                contractOrder.setStatus(ContractOrderStatus.TRADING);
//                            }else{
//                                contractOrder.setStatus(ContractOrderStatus.TRADING);
//                            }
//
//                        }
//
//
//                    }
//                }
//            }
//
//
//        }
//    }



    @KafkaListener(topics = "contract-order-cancel",containerFactory = "kafkaListenerContainerFactory")
    public void onOrderCancel(List<ConsumerRecord<String,String>> records){
        for (int i = 0; i < records.size(); i++) {
            ConsumerRecord<String, String> record = records.get(i);
            log.info("onOrderCancel:topic={},accessKey={}", record.topic(), record.key());
            String symbol = record.key();
            ContractOrder order = JSON.parseObject(record.value(), ContractOrder.class);
            ContractOrder resultOrder = order;
            if (order == null) {
                return;
            }
            CoinTrader trader = traderFactory.getTrader(symbol);
            if (trader.getReady()) {
                try {
                    try {
                        MessageResult result = trader.cancelOrder(order);
                        log.info("onOrderCancel:result =========================== = {}", result);
                        if (result.getCode()==0) {
                            kafkaTemplate.send("contract-simple-order-cancel-success", order.getSymbol(), JSON.toJSONString(resultOrder));
                        }
                    } catch (Exception e) {
                        log.error("====取消订单出错===e={}",e);
                    }
                } catch (Exception e) {
                    log.error("====取消订单出错===e={}",e);
                }
            }
        }
    }

    public void rate() {

    }


}
